import com.aliyun.odps.data.Record;
import com.aliyun.odps.data.TableInfo;
import com.aliyun.odps.mapred.JobClient;
import com.aliyun.odps.mapred.MapperBase;
import com.aliyun.odps.mapred.ReducerBase;
import com.aliyun.odps.mapred.conf.JobConf;
import com.aliyun.odps.mapred.utils.InputUtils;
import com.aliyun.odps.mapred.utils.OutputUtils;
import com.aliyun.odps.mapred.utils.SchemaUtils;

import java.io.IOException;
import java.text.DateFormat;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Iterator;

public class Main {
    public static class TokenizerMapper extends MapperBase {
        private Record key;
        private Record value;
        @Override
        public void setup(TaskContext context) throws IOException {
            key= context.createMapOutputKeyRecord();
            value = context.createMapOutputValueRecord();
        }
        @Override
        public void map(long recordNum, Record record, TaskContext context)
                throws IOException {
            VIN vin1 = new VIN(record.getString("vin"));
            String vin= vin1.getResult().split(":")[0];
            String province = util.province_notnull(record.getString("province"));

            String fault_description = new Res_fault().code_descroption(record.getString("fault_codes"));
            String[] fault_descriptionArr = fault_description.split("QAZ123");
            Long length=0L;
            if(fault_descriptionArr.length>1){
                length=1L;
            }






            key.set("vin",vin);

            value.set("sum1",length);
            value.set("province",province);

            context.write(key, value);
        }
    }


    public static class SumCombiner extends ReducerBase {
        private Record count;
        @Override
        public void setup(TaskContext context) throws IOException {
            count = context.createMapOutputValueRecord();
        }
        @Override
        public void reduce(Record key, Iterator<Record> values, TaskContext context)
                throws IOException {
            long c = 0;
            String province = "空";

            while (values.hasNext()) {

                Record val = values.next();
                c += (Long) val.get(0);
                province=(String)val.get(1);






            }
            count.set(0, c);
            count.set(1, province);
            context.write(key, count);
        }
    }

    public static class SumReducer extends ReducerBase {
        private Record result = null;
        @Override
        public void setup(TaskContext context) throws IOException {
            result = context.createOutputRecord();
        }
        @Override
        public void reduce(Record key, Iterator<Record> values, TaskContext context)
                throws IOException {
            long c = 0;
            String province = "空";

            while (values.hasNext()) {
                Record val = values.next();
                c += (Long) val.get(0);
                province = (String) val.get(1);





            }
            result.set(0, key.get(0));
            result.set(1, c);
            result.set(2, province);


            context.write(result);
        }
    }
    public static void main(String[] args) throws Exception {
        if (args.length != 2) {
            System.err.println("Usage: WordCount <in_table> <out_table>");
            System.exit(2);
        }
        JobConf job = new JobConf();
        job.setMapperClass(Main.TokenizerMapper.class);
        job.setCombinerClass(Main.SumCombiner.class);
        job.setReducerClass(Main.SumReducer.class);
        job.setMapOutputKeySchema(SchemaUtils.fromString("vin:string"));
        job.setMapOutputValueSchema(SchemaUtils.fromString("sum1:bigint,province:string"));
        InputUtils.addTable(TableInfo.builder().tableName(args[0]).build(), job);
        OutputUtils.addTable(TableInfo.builder().tableName(args[1]).build(), job);
        JobClient.runJob(job);
    }
}
